#include <stdio.h>
#include <unistd.h>
#include <iostream>
#include <queue>

#define THREAD_NUM 2

//线程安全的队列
class RingQueue
{
    public:
        RingQueue()
        {
            capacity_ = 10;
            pthread_mutex_init(&lock_, NULL);
            pthread_cond_init(&cons_, NULL);
            pthread_cond_init(&prod_, NULL);
        }

        ~RingQueue()
        {
            pthread_mutex_destroy(&lock_);
            pthread_cond_destroy(&cons_);
            pthread_cond_destroy(&prod_);
        }

        //生产者线程调用
        void Push(int data)
        {
            pthread_mutex_lock(&lock_);
            while(que_.size() >= capacity_)
            {
                pthread_cond_wait(&prod_, &lock_);
            }
            que_.push(data);
            pthread_mutex_unlock(&lock_);

            pthread_cond_signal(&cons_);
        }

        //消费者线程使用
        void Pop(int* data)
        {
            pthread_mutex_lock(&lock_);
            while(que_.empty())
            {
                pthread_cond_wait(&cons_, &lock_);
            }
            *data = que_.front();
            que_.pop();
            printf("i am consumer %p, i consume %d\n", pthread_self(), *data);
            pthread_mutex_unlock(&lock_);

            pthread_cond_signal(&prod_);
        }
    private:
        std::queue<int> que_;
        size_t capacity_;
        pthread_mutex_t lock_;

        pthread_cond_t cons_;
        pthread_cond_t prod_;
};

void* ConsumeStart(void* arg)
{
    RingQueue* rq = (RingQueue*)arg;
    while(1)
    {
        int data;
        rq->Pop(&data);
    }
    return NULL;
}

int g_data = 1;
pthread_mutex_t g_lock;

void* ProductStart(void* arg)
{
    RingQueue* rq = (RingQueue*)arg;
    while(1)
    {
        rq->Push(g_data);
        //lock
        pthread_mutex_lock(&g_lock);
        g_data++;
        //unlock
        pthread_mutex_unlock(&g_lock);
        sleep(1);
    }
    return NULL;
}



int main()
{
    pthread_mutex_init(&g_lock, NULL);
    RingQueue* rq = new RingQueue();
    if(rq == NULL)
    {
        return 0;
    }

    pthread_t cons[THREAD_NUM], prod[THREAD_NUM], prod1[THREAD_NUM];
    for(int i = 0; i < THREAD_NUM; i++)
    {
        int ret = pthread_create(&cons[i], NULL, ConsumeStart, (void*)rq);
        if(ret < 0)
        {
            perror("pthread_create");
            return 0;
        }
        ret = pthread_create(&prod[i], NULL, ProductStart, (void*)rq);
        if(ret < 0)
        {
            perror("pthread_create");
            return 0;
        }
        ret = pthread_create(&prod1[i], NULL, ProductStart, (void*)rq);
        if(ret < 0)
        {
            perror("pthread_create");
            return 0;
        }
    }

    for(int i = 0; i < THREAD_NUM; i++)
    {
        pthread_join(cons[i], NULL);
        pthread_join(prod[i], NULL);
        pthread_join(prod1[i], NULL);
    }

    pthread_mutex_destroy(&g_lock);
    return 0;
}
